package com.xctech.yace.thread;

import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.List;
import java.util.concurrent.Future;

@Data
@NoArgsConstructor
public class ProducerssThread implements Runnable{

    private KafkaProducer<String,String> producer;

    private List<ProducerRecord<String,String>> records;

    public ProducerssThread(KafkaProducer<String, String> producer, List<ProducerRecord<String, String>> records) {
        this.producer = producer;
        this.records = records;
    }

    @Override
    public void run() {

        String threadName = Thread.currentThread()+"|"+
                Thread.currentThread().getId()+"|"+Thread.currentThread().getName();

        for(int i = 0;i < records.size();i++){
            producer.send(records.get(i), (metadata, e) -> {
                if (null != e) {
                    e.printStackTrace();
                }
                if (null != metadata) {
                    System.out.println(threadName +
                            "-消息发送成功 ： " + String.format("offset: %s, partition:%s, topic:%s timestamp:%s",
                            metadata.offset(), metadata.partition(), metadata.topic(), metadata.timestamp()));
                }
            });
        }
    }
}
